package worker;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Properties;
import java.util.ResourceBundle;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import server.Content;

import com.mongodb.BasicDBObject;
import com.mongodb.DBCollection;
import com.mongodb.DBObject;

import data.Data;

public class WorkerContent implements MessageListener, ExceptionListener {

	private static ActiveMQConnectionFactory connectionFactory;
	private static Connection connection=null;
	private Session session;
	private Destination destination;
	private boolean transacted = false;
	public static ArrayList<WorkerContent> listWorker = new ArrayList<WorkerContent>();

	public WorkerContent() throws IOException {
		try {
			setUp();
			try {
				createConsumerAndReceiveAMessage();

				if (!listWorker.contains(this)) {
					listWorker.add(this);
				}

			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	private void setUp() throws JMSException, IOException {
		ResourceBundle resource = ResourceBundle.getBundle("config");
		connectionFactory = new ActiveMQConnectionFactory(
				ActiveMQConnection.DEFAULT_USER,
				ActiveMQConnection.DEFAULT_PASSWORD, "failover:tcp://"+resource.getString("server")+":61616");
		if(connection == null)
        {
			connection = connectionFactory.createConnection();
			connection.start();
        }
		session = connection
				.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
		
		destination = session.createQueue(resource.getString("queue"));
	}

	private void createConsumerAndReceiveAMessage() throws JMSException,
			InterruptedException {

		MessageConsumer consumer = session.createConsumer(destination);

		connection.setExceptionListener(this);
		consumer.setMessageListener(this);
	}

	public void cleanUp() {
		try {
			this.connection.close();
			listWorker.remove(this);

		} catch (JMSException e) {
			e.printStackTrace();
		}

	}

	public void onMessage(Message message) {
		if (message instanceof TextMessage) {
			TextMessage textMessage = (TextMessage) message;
			try {
				System.out
						.println("Received message: " + textMessage.getText());

				String link = textMessage.getText().toLowerCase();

				DBCollection coll = Data.data("urlcraw");
				DBObject url = coll.findOne(new BasicDBObject("url",link));
				if(url != null)
				{
					BasicDBObject obj = new BasicDBObject();
					obj.put("url", link);
	
					Thread[] group = new Thread[1];
					try {
						group[0] = new Content(obj,url);
					} catch (Exception e) {
						e.printStackTrace();
					}
	
					new InsertContent(25000, group, obj);
				}

			} catch (JMSException ex) {
				System.out.println("Error reading message: " + ex);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		} else {
			System.out.println("Received: " + message);
		}
	}

	synchronized public void onException(JMSException ex) {
		this.cleanUp();
	}

	public static void main(String[] args) throws IOException {
		new WorkerContent();
	}
}
